use crate::cache::get_mysql_pool;
#[cfg(feature = "postgres")]
use crate::cache::get_pg_pool;
use crate::keys::VddKey;
use crate::record_db::{self, RecordDbTrait};
use crate::sqlx::SqlxDefineTrait;
use crate::{cache, run_by_store, subscribe, user_db, value_db, visit_db};
use serde_json::Value;
use std::sync::Arc;
use visit_dd_core::i18n_text::E_ARG_MISS;
use visit_dd_core::model::{DataStoreEngine, VddSchema};
use visit_dd_core::proj::APP_CONFIG;
use visit_dd_core::query::{QueryDataPage, QueryPlan, ShareQueryParams, WebQueryParams};
use visit_dd_core::record::{VddDisplay, VddRecord};
use visit_dd_core::value::VddValue;
use visit_dd_core::{i18n, Ares, Astr, FieldId, RecordId, UserId, AC_ADD, AC_EDIT};

///
pub(crate) async fn select_records_in_index(
    sqp: &ShareQueryParams,
    index: &[u64],
) -> QueryDataPage<VddRecord> {
    let visitor = user_db::u_load(&sqp.vg.u_id, false, false).await;
    let mut record_vec = Vec::<Arc<VddRecord>>::new();
    let mut last_index = sqp.params.last_id;
    let fves = sqp
        .params
        .filters
        .iter()
        .filter(|qp| qp.as_vdd_fve().is_some())
        .map(|qp| qp.as_vdd_fve().unwrap())
        .collect::<Vec<_>>();
    let mut ids = Vec::from(index);
    if !fves.is_empty() {
        // 有设定字段比对查询条件
        let rec_ids = value_db::select_rec_ids_by_query(&fves)
            .iter()
            .map(|rid| rid.0)
            .collect::<Vec<u64>>();
        ids.retain(|id| rec_ids.contains(id));
    }
    // tracing::warn!("{:?}", ids);
    for id in ids.iter().skip(last_index as usize) {
        last_index += 1;
        let record = record_db::load_record(&sqp.schema, &RecordId(*id), &[]);
        if let Some(mut record) = record {
            if let Ok((can_edit, can_delete)) = sqp.can_accept(&mut record) {
                record
                    .change_by_vg_for_read(&sqp.vg.rule_f, &visitor.get_pub_key())
                    .unwrap();
                // 将记录对某个访问者的权限写入记录中
                record
                    .value
                    .insert(FieldId(11), Arc::new(VddValue::S211(can_edit)));
                record
                    .value
                    .insert(FieldId(12), Arc::new(VddValue::S211(can_delete)));
                record_vec.push(Arc::new(record));
            }
        }
        if record_vec.len() >= sqp.params.page_size as usize {
            break;
        }
    }
    QueryDataPage {
        data: record_vec,
        all_record_count: index.len(),
        last_index,
    }
}

/// 模型作者查询自己的模型记录
pub async fn page(p: &WebQueryParams) -> QueryDataPage<VddDisplay> {
    // tracing::info!("{:?}", p);
    let index_key = VddKey::IndexModel2Record(p.model_id.unwrap()).to_string();
    let mut ids = cache::get_index(&index_key);
    if p.new_to_old {
        ids.reverse();
    }
    // tracing::info!("ids: {:?}", ids);
    let mut record_vec = Vec::<Arc<VddDisplay>>::new();
    let mut last_index = p.last_id;
    let schema = cache::get_schema(p.model_id.expect(i18n!(E_ARG_MISS).as_ref()))
        .await
        .unwrap();
    let fves = p
        .filters
        .iter()
        .filter(|qp| qp.as_vdd_fve().is_some())
        .map(|qp| qp.as_vdd_fve().unwrap())
        .collect::<Vec<_>>();
    if !fves.is_empty() {
        // 有设定字段比对查询条件
        let rec_ids = value_db::select_rec_ids_by_query(&fves)
            .iter()
            .map(|rid| rid.0)
            .collect::<Vec<u64>>();
        ids.retain(|id| rec_ids.contains(id));
    }
    // tracing::warn!("{:?}", ids);
    for rec_id in ids.iter().skip(last_index as usize) {
        last_index += 1;
        // tracing::info!("record id is {:?}", rec_id);
        let record = record_db::load_record(&schema, &RecordId(*rec_id), &[]);
        if record.is_some() {
            let record = record.unwrap();
            if p.can_accept(&record.meta).is_ok() {
                // tracing::info!("record is accepted");
                record_vec.push(record.new_display(&schema).await.into());
            }
        }
        if record_vec.len() >= p.page_size as usize {
            break;
        }
    }
    QueryDataPage {
        data: record_vec,
        all_record_count: ids.len(),
        last_index,
    }
}

/// 解析参数 并从数据库中查找对应的访问权限
pub async fn sqp_from_json(j: &Value, uid: &UserId) -> Result<ShareQueryParams, Astr> {
    let model_id = j["modelId"].as_u64().unwrap();
    let schema = cache::get_schema(model_id).await?;
    let vg = visit_db::vg_try_load_in_model_by_user(schema.model.id, uid).await;
    if vg.is_none() {
        return Err(i18n!(1101));
    }
    let vg = vg.unwrap();
    let params = WebQueryParams::try_from(j)?;
    Ok(ShareQueryParams {
        schema,
        vg: vg.into(),
        params: params.into(),
    })
}

async fn rec_to_display_page(
    page: &QueryDataPage<VddRecord>,
    schema: &VddSchema,
) -> QueryDataPage<VddDisplay> {
    let mut recordd_vec = Vec::<Arc<VddDisplay>>::new();
    for rec in page.data.iter() {
        recordd_vec.push(rec.new_display(schema).await.into());
    }
    QueryDataPage {
        data: recordd_vec,
        all_record_count: page.all_record_count,
        last_index: page.last_index,
    }
}

/// 访问者在平台查询模型记录
pub async fn share_page(sqp: &ShareQueryParams) -> QueryDataPage<VddDisplay> {
    let index_key = VddKey::IndexModel2Record(sqp.schema.model.id);
    let mut ids = cache::get_index(&index_key.to_string());
    if sqp.params.new_to_old {
        ids.reverse();
    }
    let page = select_records_in_index(sqp, &ids).await;
    rec_to_display_page(&page, &sqp.schema).await
}

/// 获取订阅更新里的记录
pub async fn subscribe_get_page(sqp: &ShareQueryParams) -> QueryDataPage<VddDisplay> {
    let sc = subscribe::sc_load(sqp.schema.model.id, &sqp.vg.u_id).await;
    match sc {
        Ok(sc2) => {
            let events = subscribe::select_events(&sc2);
            let ids = events
                .iter()
                .map(|event| {
                    if event.action == AC_ADD || event.action == AC_EDIT {
                        Some(event.rid.0)
                    } else {
                        None
                    }
                })
                .filter(|event| event.is_some())
                .map(|event| event.unwrap())
                .collect::<Vec<_>>();
            let page = select_records_in_index(sqp, &ids).await;
            rec_to_display_page(&page, &sqp.schema).await
        }
        Err(..) => QueryDataPage {
            all_record_count: 0,
            last_index: 0,
            data: Vec::<Arc<VddDisplay>>::new(),
        },
    }
}

pub async fn load_plan(u_id: &UserId, id: u64) -> QueryPlan {
    run_by_store!(impl_rdb::load_plan, impl_sqlx::load_plan, u_id, id)
}

pub async fn select_plans(u_id: &UserId) -> Vec<Arc<QueryPlan>> {
    run_by_store!(impl_rdb::select_plans, impl_sqlx::select_plans, u_id)
}

pub async fn delete_plan(u_id: &UserId, id: u64) -> Ares {
    run_by_store!(impl_rdb::delete_plan, impl_sqlx::delete_plan, u_id, id)
}

pub async fn save_plan(plan: &QueryPlan, is_update: bool) -> Ares {
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::save_plan(plan, is_update)
    } else {
        let ds = APP_CONFIG.meta_store.clone().unwrap();
        match ds.store_engine {
            #[cfg(feature = "postgres")]
            DataStoreEngine::Postgres => {
                get_pg_pool(ds.as_ref())
                    .await
                    .save_plan(plan, is_update)
                    .await
            }
            DataStoreEngine::Mysql => {
                get_mysql_pool(ds.as_ref())
                    .await
                    .save_plan(plan, is_update)
                    .await
            }
        }
    }
}

mod impl_rdb {
    use rust_rocksdb::WriteBatchWithTransaction;
    use serde_json::json;
    use std::sync::Arc;
    use visit_dd_core::i18n_text::E_ROCKSDB;
    use visit_dd_core::{i18n, to_sha256_hash};
    use visit_dd_core::{query::QueryPlan, Ares, UserId};

    use crate::delete_by_value;
    use crate::{keys::VddKey, load_by_key_head, ROCKSDB};

    pub(super) fn load_plan(u_id: &UserId, id: u64) -> QueryPlan {
        let vk = VddKey::BoundQueryPlan(u_id.clone(), id);
        let data = ROCKSDB.get_cf(vk.db_name().unwrap(), vk.to_string());
        if let Ok(Some(vu8)) = data {
            serde_json::from_slice::<QueryPlan>(&vu8).unwrap()
        } else {
            panic!("{}", i18n!(E_ROCKSDB))
        }
    }

    pub(super) fn select_plans(u_id: &UserId) -> Vec<Arc<QueryPlan>> {
        let vk = VddKey::BoundQueryPlan(u_id.clone(), 0);
        load_by_key_head(&vk)
            .iter()
            .map(|v| serde_json::from_slice::<QueryPlan>(v).unwrap())
            .map(|qp| Arc::new(qp))
            .collect::<Vec<_>>()
    }

    pub(super) fn delete_plan(u_id: &UserId, id: u64) -> Ares {
        let mut batch = WriteBatchWithTransaction::<true>::default();
        let vk = VddKey::BoundQueryPlan(u_id.clone(), id);
        batch.delete_cf(vk.db_name().unwrap(), vk.to_string());
        let vk = VddKey::UniqueQueryPlan(u_id.clone(), "".into());
        delete_by_value(&mut batch, &vk, &id.to_le_bytes());
        ROCKSDB.write(batch).map_err(|e| e.to_string().into())
    }

    pub(super) fn save_plan(plan: &QueryPlan, is_update: bool) -> Ares {
        let hash = to_sha256_hash(plan.name.as_bytes());
        let vk_u = VddKey::UniqueQueryPlan(plan.u_id.clone(), hash);
        if !is_update {
            if let Ok(Some(_)) = ROCKSDB.get_cf(vk_u.db_name().unwrap(), vk_u.to_string()) {
                return Err("已有同名称查询计划存在".into());
            }
        }
        let vk = VddKey::BoundQueryPlan(plan.u_id.clone(), plan.id);
        let mut batch = WriteBatchWithTransaction::<true>::default();
        batch.put_cf(
            vk.db_name().unwrap(),
            vk.to_string(),
            json!(plan).to_string(),
        );
        batch.put_cf(
            vk_u.db_name().unwrap(),
            vk_u.to_string(),
            plan.id.to_le_bytes(),
        );
        ROCKSDB.write(batch).map_err(|e| e.to_string().into())
    }
}

mod impl_sqlx {
    use std::sync::Arc;
    use visit_dd_core::i18n_text::E_ROCKSDB;
    use visit_dd_core::{i18n, Ares};
    use visit_dd_core::{query::QueryPlan, UserId};

    use crate::{keys::VddKey, sqlx::SqlxDefineTrait};

    pub(super) async fn load_plan(
        pool: &impl SqlxDefineTrait,
        u_id: &UserId,
        id: u64,
    ) -> QueryPlan {
        let vk = VddKey::BoundQueryPlan(u_id.clone(), id);
        let data = pool.select_bounds(vk).await;
        let data = data.first();
        if let Some(j) = data {
            serde_json::from_str::<QueryPlan>(&j).unwrap()
        } else {
            panic!("{}", i18n!(E_ROCKSDB))
        }
    }

    pub(super) async fn select_plans(
        pool: &impl SqlxDefineTrait,
        u_id: &UserId,
    ) -> Vec<Arc<QueryPlan>> {
        let vk = VddKey::BoundQueryPlan(u_id.clone(), 0);
        let data = pool.select_bounds(vk).await;
        data.iter()
            .map(|v| serde_json::from_str::<QueryPlan>(v).unwrap())
            .map(|qp| Arc::new(qp))
            .collect::<Vec<_>>()
    }

    pub(super) async fn delete_plan(pool: &impl SqlxDefineTrait, u_id: &UserId, id: u64) -> Ares {
        let vk = VddKey::BoundQueryPlan(u_id.clone(), id);
        pool.delete_bounds(vk).await?;
        // 偷了个懒，没有启动事务
        let vk_u = VddKey::UniqueQueryPlan(u_id.clone(), "".into());
        pool.delete_unique_by_value(vk_u, id).await
    }
}
